<?php
/**
 * Created by PhpStorm.
 * User: hedx
 * Date: 2016/3/1
 * Time: 15:10
 */

namespace common\mq\alimns;

require_once(dirname(__FILE__) . '/lib/mns-autoloader.php');

use AliyunMNS\Client;
use AliyunMNS\Exception\MnsException;
use AliyunMNS\Model\QueueAttributes;
use AliyunMNS\Model\TopicAttributes;
use AliyunMNS\Queue;
use AliyunMNS\Requests\CreateQueueRequest;
use AliyunMNS\Requests\CreateTopicRequest;
use AliyunMNS\Requests\ListQueueRequest;
use AliyunMNS\Requests\PublishMessageRequest;
use AliyunMNS\Requests\SendMessageRequest;
use AliyunMNS\Responses\DeleteQueueResponse;
use AliyunMNS\Responses\PublishMessageResponse;
use AliyunMNS\Responses\ReceiveMessageResponse;
use common\mq\MessageServiceInterface;
use Yii;
use yii\base\Component;
use yii\base\Exception;

class AliMessageService extends Component implements MessageServiceInterface
{

    /**
     * @var Client AliyunMNSClient
     */
    protected static $client;

    /**
     * 阿里云的Access Key ID
     * @var string
     */
    public $accessId;

    /**
     * 阿里云的Access Key Secret
     * @var string
     */
    public $accessKey;

    /**
     * 阿里云MNS的接入地址：不同地址的同名队列是各自独立的
     * @var string
     */
    public $endPoint;

    public function init()
    {
        parent::init();
        if (empty($this->accessId) || empty($this->accessKey)) {
            throw new Exception("Please specify parameter 'accessId' and 'accessKey' for Ali Message Service.");
        }
        if (empty(self::$client)) {
            self::$client = new Client($this->endPoint, $this->accessId, $this->accessKey);
        }
    }

    /**
     * 返回 AliMNS Client
     *
     * @return Client
     */
    public function getClient()
    {
        return self::$client;
    }

    /**
     * 创建一个队列，支持一对一消息的发送和接收
     *
     * @param string $queueName 队列名称
     * @param QueueAttributes $queueAttributes 队列属性
     * @return boolean
     */
    public function createQueue($queueName, $queueAttributes = NULL)
    {
        if (empty($queueAttributes)) {
            $queueAttributes = new QueueAttributes();
            $queueAttributes->setMessageRetentionPeriod(345600); //消息在该Queue中最长的存活时间，默认设置4天
            $queueAttributes->setVisibilityTimeout(5); //消息由inactive状态转为active状态的时间，默认设置为1个小时
            $queueAttributes->setPollingWaitSeconds(30); //当Queue中没有消息时，ReceiveMessage请求最长的等待时间
        }
        $request = new CreateQueueRequest($queueName, $queueAttributes);
        try {
            $response = $this->getClient()->createQueue($request);
            if ($response->isSucceed()) {
                return true;
            }
        } catch (MnsException $e) {
            Yii::error("创建队列 " . $queueName . " 失败，原因：" . $e);
        }
        return false;
    }

    /**
     * 根据队列名称删除一个队列
     *
     * @param string $queueName 队列名称
     * @return DeleteQueueResponse
     */
    public function deleteQueue($queueName)
    {
        return $this->getClient()->deleteQueue($queueName);
    }

    /**
     * 向指定的队列发送一条消息，如果不想消息被马上消费，可以指定延时时间 delaySeconds,
     * 消息将在指定的时间后被消费，否则消息可随时被消费。如果消息发送成功则返回一个唯一
     * 的messageId用来标识此消息。
     *
     * @param string $queueName 队列名称
     * @param string $messageBody 消息主体  {title:"",sender:"",receiver:"",body:""} json格式
     * @param int $priority 消息优先级范围：1-16 值越小优先级越高， 默认为8
     * @param int $delaySeconds 延时时间范围: 0-604800 秒，默认无延时
     * @return array ["isSuccess"=>"boolean","messageId"=>"foo","errMsg"=>"foo"]
     */
    public function sendMessage($queueName, $messageBody, $priority = 8, $delaySeconds = NULL)
    {
        $queue = $this->getClient()->getQueueRef($queueName);
        $request = new SendMessageRequest($messageBody, $delaySeconds, $priority);
        $result = ["isSuccess" => false, "messageId" => "", "errMsg" => ""];
        try {
            $response = $queue->sendMessage($request);
            if ($response->isSucceed()) {
                $result["messageId"] = $response->getMessageId();
                $result["isSuccess"] = true;
            }
        } catch (MnsException $e) {
            Yii::error("发送消息失败，原因：" . $e);
            $result["errMsg"] = $e;
        }
        return $result;
    }

    /**
     * 读取某个队列的一个消息，读取成功则会调用callback回调函数消费消息，消息消费后将被删除
     *
     * @param Queue $queue 队列
     * @param array $callback 回调函数
     * @param array $parameters 兼容其他mq实现时的参数对象
     * @return mixed
     */
    public function receiveMessage($queue, $callback, $parameters = [])
    {
        $response = null;
        try {
            $response = $queue->receiveMessage();
            //消息接收后的处理
            $this->handleMessageReceive($queue, $response, $callback);
        } catch (MnsException $e) {
            //Yii::error("消息接收失败，原因：" . $e);
        } catch (\Exception $e) {
            Yii::error("消息消费失败，原因：" . $e);
            $this->handleCallbackException($queue, $response, $e);
        }
        return null;
    }

    /**
     * 监听某个队列，如果队列中有可消费的消息则调用callback回调函数消费消息，消费完后在队列中删除该消息
     *
     * @param string $queueName 队列名称
     * @param array $callback 回调函数数组 => ["object","callback"]
     * @param array $parameters 兼容其他mq实现时的参数对象
     * @return mixed
     */
    public function listen($queueName, $callback, $parameters = [])
    {
        $queue = $this->getClient()->getQueueRef($queueName);
        $response = null;
        while (true) {
            try {
                $response = $queue->receiveMessage();
                //消息接收后的处理
                $this->handleMessageReceive($queue, $response, $callback);
            } catch (MnsException $e) {
                break; //队列中无消息时直接退出
            } catch (\Exception $e) {
                Yii::error("消息消费失败，原因：" . $e);
                $this->handleCallbackException($queue, $response, $e);
            }
        }
        return null;
    }

    /**
     * 监听某个队列，如果队列中有可消费的消息则调用callback回调函数消费消息，消费完后在队列中删除该消息
     *
     * @param string $queueName 队列名称
     * @param array $callback 回调函数数组 => ["object","callback"]
     * @param array $parameters 兼容其他mq实现时的参数对象
     * @param int $waitSeconds 等待秒数
     * @return mixed
     */
    public function listenNoBreak($queueName, $callback, $parameters = [], $waitSeconds = 1)
    {
        $queue = $this->getClient()->getQueueRef($queueName);
        $response = null;
        while (true) {
            try {
                $response = $queue->receiveMessage();
                //消息接收后的处理
                $this->handleMessageReceive($queue, $response, $callback);
            } catch (MnsException $e) {
                sleep($waitSeconds);
            } catch (\Exception $e) {
                Yii::error("消息消费失败，原因：" . $e);
                $this->handleCallbackException($queue, $response, $e);
            }
        }
        return null;
    }


    /**
     * 处理消息的接收
     *
     * @param Queue $queue
     * @param ReceiveMessageResponse $response
     * @param $callback
     */
    private function handleMessageReceive($queue, $response, $callback)
    {
        if ($response->isSucceed()) {
            try {
                //通过回调函数消费消息
                call_user_func($callback, $response);
            } catch (\Exception $exception) {
                //如果消费失败则将消息内容插入异常队列
                $messageBody = $response->getMessageBody();
                $errMsg = $exception->getMessage();
                $newMessageBody = json_encode(["messageBody" => $messageBody, "errMsg" => $errMsg]);
                $this->sendMessage('exceptionQueue', $newMessageBody);
            }

            //消息调用完毕后删除消息
            $receiptHandle = $response->getReceiptHandle();
            $res = $queue->deleteMessage($receiptHandle);
        }
    }

    /**
     * 处理回调函数中抛出的异常
     *
     * @param Queue $queue
     * @param ReceiveMessageResponse $response
     * @param string $e
     */
    private function handleCallbackException($queue, $response, $e)
    {
        if (empty($response) || !($response instanceof ReceiveMessageResponse)) {
            return;
        }
        //$queueName = $queue->getQueueName();
        //$messageId = $response->getMessageId();

        //记录失败原因
        /*$messageReceiveRecord = MessageHelper::getMessageReceive($queueName, $messageId);
        if (!empty($messageReceiveRecord)) {
            $messageReceiveRecord->remark = "尝试第" . $response->getDequeueCount() . "次投递消息时失败，原因:" . $e;
            $messageReceiveRecord->update();
        }*/

        //如果接收此消息大于等于1次则在阿里云队列中删除此消息，不再接收
        if ($response->getDequeueCount() >= 1) {
            $receiptHandle = $response->getReceiptHandle();
            $res = $queue->deleteMessage($receiptHandle);
            /*if ($res->isSucceed()) {
                MessageHelper::updateMessageReceiveDeleteStatus($queueName, $messageId, MessageServiceReceiveModel::IS_DELETE);
            }*/
        }
    }

    /**
     * 创建一个主题，主题模型提供pub/sub模型，支持一对多消息的订阅和通知
     *
     * @param string $topicName 主题名称
     * @param TopicAttributes $topicAttributes 主题属性
     * @return boolean
     */
    public function createTopic($topicName, $topicAttributes = NULL)
    {
        if (empty($topicAttributes)) {
            $topicAttributes = new TopicAttributes();
            $topicAttributes->setMaximumMessageSize(65536); //消息体最大值，默认64KB
        }
        $request = new CreateTopicRequest($topicName, $topicAttributes);
        try {
            $response = $this->getClient()->createTopic($request);
            if ($response->isSucceed()) {
                return true;
            }
        } catch (MnsException $e) {
            Yii::error("创建主题 " . $topicName . " 失败，原因：" . $e);
        }
        return false;
    }

    /**
     * 删除一个主题，一旦删除成功，主题内所有消息也一并删除，不可恢复
     *
     * @param $topicName
     * @return mixed
     */
    public function deleteTopic($topicName)
    {
        return $this->getClient()->deleteTopic($topicName);
    }

    /**
     * 向指定的topic发布一个订阅消息，此消息会被所有订阅了此topic的订阅者接收
     *
     * @param $topicName
     * @param $messageBody
     * @return mixed
     */
    public function publishMessage($topicName, $messageBody)
    {
        $topic = $this->getClient()->getTopicRef($topicName);
        $request = new PublishMessageRequest($messageBody);
        $result = ["isSuccess" => false, "messageId" => "", "errMsg" => ""];
        try {
            /** @var PublishMessageResponse $response */
            $response = $topic->publishMessage($request);
            if ($response->isSucceed()) {
                $result["messageId"] = $response->getMessageId();
                $result["isSuccess"] = true;
            }
        } catch (MnsException $e) {
            Yii::error("发送消息失败，原因：" . $e);
            $result["errMsg"] = $e;
        }
        return $result;
    }

    /**
     * 获取所有的队列数组
     *
     * @return mixed
     */
    public function getAllQueueList()
    {
        $listQueueRequest = new ListQueueRequest();
        $response = $this->getClient()->listQueue($listQueueRequest);
        if ($response->getStatusCode() == '200') {
            return $response->getQueueNames();
        }
        return [];
    }
}